home
***
CD-ROM
|
disk
|
FTP
|
other
***
search
/
Cream of the Crop 26
/
Cream of the Crop 26.iso
/
os2
/
pvm34b3.zip
/
pvm34b3
/
pvm3
/
src
/
lpvm.c
< prev
next >
Wrap
Text File
|
1997-08-08
|
81KB
|
3,567 lines
static char rcsid[] =
"$Id: lpvm.c,v 1.30 1997/06/27 18:04:17 pvmsrc Exp $";
/*
* PVM version 3.4: Parallel Virtual Machine System
* University of Tennessee, Knoxville TN.
* Oak Ridge National Laboratory, Oak Ridge TN.
* Emory University, Atlanta GA.
* Authors: J. J. Dongarra, G. E. Fagg, M. Fischer
* G. A. Geist, J. A. Kohl, R. J. Manchek, P. Mucci,
* P. M. Papadopoulos, S. L. Scott, and V. S. Sunderam
* (C) 1997 All Rights Reserved
*
* NOTICE
*
* Permission to use, copy, modify, and distribute this software and
* its documentation for any purpose and without fee is hereby granted
* provided that the above copyright notice appear in all copies and
* that both the copyright notice and this permission notice appear in
* supporting documentation.
*
* Neither the Institutions (Emory University, Oak Ridge National
* Laboratory, and University of Tennessee) nor the Authors make any
* representations about the suitability of this software for any
* purpose. This software is provided ``as is'' without express or
* implied warranty.
*
* PVM version 3 was funded in part by the U.S. Department of Energy,
* the National Science Foundation and the State of Tennessee.
*/
/*
* lpvm.c
*
* Libpvm core for unix environment.
*
$Log: lpvm.c,v $
* Revision 1.30 1997/06/27 18:04:17 pvmsrc
* Integrated WIN32 changes.
*
* Revision 1.29 1997/06/12 20:10:37 pvmsrc
* Made sure all communications for TC_* task control messages
* use the SYSCTX_TC system context.
* - some messages being sent in default context... D-Oh...
*
* Revision 1.28 1997/05/07 21:19:23 pvmsrc
* swapped logic around SOCKLENISINT so that it is now
* the default path and SOCKLENISUINT is the compile-time
* selected path.
*
* Now, AIX 4.1 will have to remove SOCKLENISUINT from
* their AIXxxx.def configure files.
*
* Revision 1.27 1997/05/07 18:37:13 pvmsrc
* AIX 3.2 vs 4.1 vs 4.2 problems resolved (I hope)
* Define oslen as size_t with compile flag SOCKLENISINT
* for AIX 4.1 to explicitly set to int.
* To use flag - set in conf/AIX46K.def.
* Tested with cc and gcc on all 3 OS versions.
*
* Revision 1.26 1997/05/05 15:19:40 pvmsrc
* context was not being set in pvmmcast.
*
* Revision 1.25 1997/05/01 14:11:46 pvmsrc
* SGI Compiler Warning Cleanup.
*
* Revision 1.24 1997/04/30 21:26:00 pvmsrc
* SGI Compiler Warning Cleanup.
*
* Revision 1.23 1997/04/29 20:29:12 pvmsrc
* Support new calling sequences defined in mppmsg.h
*
* Revision 1.22 1997/04/25 19:21:18 pvmsrc
* Handle inplace bodies of zero length correctly
*
* Revision 1.21 1997/04/24 21:06:59 pvmsrc
* Excised unused variables.
* Support for pvm_psend/precv constructs
*
* Revision 1.20 1997/04/21 14:58:28 pvmsrc
* Changed #ifdefs that checked IMA_RS6K,IMA_SP2MPI & IMA_AIX46K
* to see if select.h was needed into single define NEEDSSELECTH.
* New archs need to set this in conf/
*
* Revision 1.19 1997/04/07 21:09:17 pvmsrc
* pvm_addmhf() - new paramter interface
*
* Revision 1.18 1997/04/03 19:23:13 pvmsrc
* Added context for TM_xxx messages
*
* Revision 1.17 1997/04/01 21:28:11 pvmsrc
* Damn Damn Damn.
* - pvm_recvinfo() returns a bufid, not an index. Damn.
*
* Revision 1.16 1997/04/01 20:48:17 pvmsrc
* Fixed tracer mbox usage:
* - pvm_getinfo() -> pvm_recvinfo(), new semantics handled (recvinfo
* sets rbuf implicitly, a la pvm_recv, need to save rbuf).
*
* Revision 1.15 1997/03/27 19:55:27 pvmsrc
* Fixed up pvmbeatask() to go get tracer info if spawned from shell:
* - env var info including trace mask, trace buffer size, trace opts.
* - use PVMTRACERCLASS mbox entry to fill in values, if matches
* on trctid, trcctx, and trctag.
*
* Revision 1.14 1997/03/07 15:12:11 pvmsrc
* Fixed annoying SGI warnings for qsort()'s int_compare() function...
*
* Revision 1.13 1997/03/06 21:50:17 pvmsrc
* Yanked out #includes for <netinet/in.h> and <netinet/tcp.h>.
* - dups with lpvm.h #includes...
*
* Revision 1.12 1997/03/06 21:06:10 pvmsrc
* - added include for lmsg.h, host.h
* - for mpp's added pvmmimd.h, mppmsg.h
* - moved struct ttpcb definition to header file lpvm.h
* - for mpps added call to mpp_ttpcb_find inside of ttpcb_find
* - mxinput rewritten to handle mpp input streams when ifdef'ed on
* - Added a stack of inplace headers instead of just one dummy header.
* so we can do asynch header transmission
* - in mxfer:
* logic written in to handle mpp streams, including bypass of
* reading routes if writing, posting a complete message for
* asynch send before checking for read routes, pass information
* to pvm_node_send (on mpps) about inplace sending so that
* pvm_node_send can be optimized for inplace (not optimized yet :-))
* - in mroute:
* direct routing socket setup turned off for MPPs
* - in pvmbeatask:
* call pvm_mpp_beatask for mpps
* don't do socket auth dance for mpps (sockets not used)
* - added routine ogm_complete to test if an outgoing message is
* complete (when sent asychronously). Returns TRUE for workstations
* since writes are synchronous
*
* Revision 1.11 1997/01/28 19:26:21 pvmsrc
* New Copyright Notice & Authors.
*
* Revision 1.10 1996/12/18 22:27:43 pvmsrc
* Extracted duplicate versions of routines from lpvm/mimd/shmem.c,
* inserted into shared lpvmgen.c:
* - pvmbailout().
* - pvmlogerror().
* - vpvmlogprintf(), pvmlogprintf(). (hope these work on MPP & shmem)
* - pvmlogperror().
*
* Revision 1.9 1996/10/25 13:57:20 pvmsrc
* Replaced old #includes for protocol headers:
* - <pvmsdpro.h>, "ddpro.h", "tdpro.h"
* With #include of new combined header:
* - <pvmproto.h>
*
* Revision 1.8 1996/10/24 22:44:31 pvmsrc
* Modified for New Tracing Facility:
* - moved #include "global.h" below other #include's for typing.
* - removed extra #include <pvm3.h> in lpvm.c...
* - added #include of new "lpvm.h" to replace explicit externs.
* - removed common control message handlers from lpvm.c:
* * extracted to lpvmgen.c for general usage.
* * pvm_tc_noop(), pvm_tc_settmask(), pvm_tc_siblings().
* -> lpvmmimd.c & lpvmshmem.c still need remainder of pvmmctl()
* replaced with control message handlers.
* - arg typing hassles with int_compare() / qsort() exacerbated...
* - modified pvmbeatask():
* * handle new tracing info, unpack tracing and output collection
* parameters into temp storage, and then check for local task
* override before applying.
* * read in new tracing env vars PVMTRCBUF & PVMTRCOPT.
* * install new common message handlers.
* * call new tev_init() routine to set up tracing stuff.
* * use new Pvmtracer structures (pvmtrc & pvmctrc) to store info.
* - removed pvm_getopt() & pvm_setopt() -> moved to common lpvmgen.c.
* - removed old tev_begin(), tev_fin() & tev_do_trace() routines.
* - updated trace event generation for pvm_getfds(), pvm_start_pvmd(),
* pvm_precv(), pvm_psend().
*
* Revision 1.7 1996/10/14 19:17:06 pvmsrc
* Used ARCHFLAG SOCKLENISUINT where socket length is unsigned int
*
* Revision 1.6 1996/10/09 21:48:19 pvmsrc
* Problem:
* --------
* The problem was the result of AIX4.2 changing the expected datatype of
* the XxxxLength parameter in the following system function calls:
*
* int bind (Socket, Name, NameLength)
* int getsockname (Socket, Name, NameLength)
* int recvfrom (Socket, Buffer, Length, Flags, From, FromLength)
* int accept (Socket, Address, AddressLength)
*
* The compiler warning message generated was:
* Function argument assignment between types "unsigned long*" and "int*"
* is not allowed.
*
* Action:
* -------
* In PVM functions where oslen was already declared_
* changed declaration from: int oslen;
* to: unsigned int oslen;
*
* In PVM functions where a local "temporary" variable such as i, j, etc.
* was used_
* added declaration of: unsigned int oslen;
* and then changed to use oslen where necessary.
*
* Revision 1.5 1996/10/04 13:17:23 pvmsrc
* New context was generated on each spawn. This is bad.
*
* Revision 1.4 1996/09/24 15:15:46 pvmsrc
* Test failure... fixed.
*
* Revision 1.3 1996/09/24 15:12:19 kohl
* Test check in...
*
* Revision 1.2 1996/09/23 23:30:53 pvmsrc
* Initial Creation - original lpvm.c.
*
* Revision 1.32 1995/11/02 16:07:10 manchek
* added NEEDSENDIAN switch.
* must declare ptr to sys_errlist const in pvmlogperror for FREEBSD.
* free replies to control messages in mxfer
*
* Revision 1.31 1995/09/06 17:37:24 manchek
* aargh, forgot pvm_precv
*
* Revision 1.30 1995/09/06 17:30:32 manchek
* pvm_psend returns not implemented instead of bad param for string type
*
* Revision 1.29 1995/07/28 16:40:58 manchek
* wrap HASERRORVARS around errno declarations
*
* Revision 1.28 1995/07/28 16:04:05 manchek
* switch endian includes on flag, not arch name
*
* Revision 1.27 1995/07/19 21:43:04 manchek
* better mxfer logging
*
* Revision 1.26 1995/07/19 21:27:49 manchek
* use pvmnametag to give symbolic message tags
*
* Revision 1.25 1995/07/18 16:59:03 manchek
* added code to generate and check crc on each message (MCHECKSUM)
*
* Revision 1.24 1995/07/03 19:05:35 manchek
* removed POWER4 arch ifdefs
*
* Revision 1.23 1995/06/28 18:18:07 manchek
* do-nothing check_for_exit so one can be in lpvmshmem.c
*
* Revision 1.22 1995/06/19 17:35:05 manchek
* addr sometimes wasn't set in pvm_tc_conreq, causing segfault
*
* Revision 1.21 1995/06/12 15:56:14 manchek
* added PGON partition size support
*
* Revision 1.20 1995/06/02 17:19:14 manchek
* added check in mxfer() for when outgoing route closed during input
*
* Revision 1.19 1995/06/01 14:43:54 manchek
* pvm_start_pvmd ignores INT, QUIT, TSTP signals
*
* Revision 1.18 1995/05/30 17:28:08 manchek
* added ifdefs for SP2MPI arch
*
* Revision 1.17 1995/05/17 17:06:54 manchek
* added PVMTASKDEBUG envar
*
* Revision 1.16 1995/05/17 16:16:36 manchek
* use FDSETISINT.
* use umbuf_get and put instead of ALLOC.
* in mksocs, read pvmd sockaddr file only once.
* in pvmbeatask, read altpid from environment each time.
* add PvmPollTime and Type, NotImplemented.
* pvm_start_pvmd reads sockaddr from pvmd instead of sleeping on addr file,
* sets PVMSOCK envar from return.
* allows better synchronization, master host overloading.
*
* Revision 1.15 1995/02/01 21:08:46 manchek
* error 4 is now PvmOverflow.
* connect is retried in case of EINTR
*
* Revision 1.14 1994/12/20 16:28:19 manchek
* added PvmShowTids case to setopt.
* removed EOF message on pvmd socket close
*
* Revision 1.13 1994/10/15 19:07:31 manchek
* don't use fd_sets when select returns -1; will hang reading.
* cast message tags for comparison as integers
*
* Revision 1.12 1994/09/02 15:23:37 manchek
* fixed segfaults in setopt when task not connected
*
* Revision 1.11 1994/07/18 19:20:18 manchek
* fix to call write() with max 4096 length for RS6K
*
* Revision 1.10 1994/06/21 19:36:23 manchek
* forgot to ifdef tt_spath
*
* Revision 1.9 1994/06/21 18:31:31 manchek
* connect in mksocs() tries several times before giving up.
* don't setsockopt() at TCP level on Unix task-task sockets
*
* Revision 1.8 1994/06/04 21:44:57 manchek
* added unix domain sockets
*
* Revision 1.7 1994/06/03 20:38:15 manchek
* version 3.3.0
*
* Revision 1.6 1993/11/30 23:50:01 manchek
* set nodelay and snd, rcv buffer sizes on t-t sockets
*
* Revision 1.5 1993/11/30 15:51:30 manchek
* beatask complains if it can't write d-auth file (fs full?)
*
* Revision 1.4 1993/10/04 20:29:05 manchek
* mksocks() and pvm_start_pvmd() now use pvmdsockfile(), not TDSOCKNAME.
* made pvm_useruid global for pvmcruft.c
*
* Revision 1.3 1993/10/04 19:10:22 manchek
* mctl() conflicts with mctl(2) - declare static for now
*
* Revision 1.2 1993/09/16 21:36:20 manchek
* pvm_start_pvmd() was freeing the return string from pvmgetpvmd()
*
* Revision 1.1 1993/08/30 23:26:48 manchek
* Initial revision
*
*/
#ifdef WIN32
#include "pvmwin.h"
#include <process.h>
#endif
#include <stdio.h>
#ifdef NEEDMENDIAN
#include <machine/endian.h>
#endif
#ifdef NEEDENDIAN
#include <endian.h>
#endif
#ifdef NEEDSENDIAN
#include <sys/endian.h>
#endif
#ifndef WIN32
#include <rpc/types.h>
#include <rpc/xdr.h>
#include <sys/socket.h>
#else
#include "..\xdr\types.h"
#include "..\xdr\xdr.h"
#endif
#include <pvm3.h>
#ifndef NOUNIXDOM
#include <sys/un.h>
#endif
#ifdef NEEDSSELECTH
#include <sys/select.h>
#endif
#include <sys/stat.h>
#include <fcntl.h>
#ifdef SYSVSTR
#include <string.h>
#define CINDEX(s,c) strchr(s,c)
#else
#include <strings.h>
#define CINDEX(s,c) index(s,c)
#endif
#include <errno.h>
#include <signal.h>
#include <pvmproto.h>
#include "pvmalloc.h"
#include "pvmfrag.h"
#include "pmsg.h"
#include "listmac.h"
#include "tvdefs.h"
#include "bfunc.h"
#include "lpvm.h"
#include <pvmtev.h>
#include "tevmac.h"
#include "host.h"
#include "waitc.h"
#include "global.h"
#include "lmsg.h"
#ifdef IMA_MPP
#include "pvmmimd.h"
#include "mppmsg.h"
#endif
#ifndef max
#define max(a,b) ((a)>(b)?(a):(b))
#endif
#ifndef min
#define min(a,b) ((a)<(b)?(a):(b))
#endif
#ifndef TTSOCKBUF
#define TTSOCKBUF 0x8000
#endif
#ifdef NOTMPNAM
#define TMPNAMFUN(x) pvmtmpnam(x)
#define LEN_OF_TMP_NAM 32
char *pvmtmpnam();
#else /*NOTMPNAM*/
#define TMPNAMFUN(x) tmpnam(x)
#ifdef L_tmpnam
#define LEN_OF_TMP_NAM L_tmpnam
#else
#define LEN_OF_TMP_NAM 64
#endif
#endif /*NOTMPNAM*/
/***************
** Globals **
** **
***************/
char *getenv();
void hex_inadport __ProtoGlarp__ (( char *, struct sockaddr_in * ));
extern char *inadport_decimal();
extern char *inadport_hex();
char *pvmgetpvmd();
char *pvmdsockfile();
char *pvmnametag();
struct pmsg *midtobuf();
struct pmsg *umbuf_new();
char *debug_flags();
struct msgid *pvm_inprecv = (struct msgid *) NULL;
#ifndef HASERRORVARS
extern int errno; /* from libc */
extern char *sys_errlist[];
extern int sys_nerr;
#endif
#if defined(IMA_PGON)
int pvmpgonpartsize = -1; /* FE prog, no partition */
#endif
/***************
** Private **
** **
***************/
static int mxfersingle = 1; /* mxfer returns after single frag */
static struct sockaddr_in pvmourinet; /* our host ip addr */
static struct ttpcb *ttlist = 0; /* dll of connected tasks */
static struct ttpcb *topvmd = 0; /* default route (to pvmd) */
static int pvmnfds = 0; /* 1 + highest bit set in fds */
static fd_set pvmrfds; /* rd fdset for mxfer() */
static struct pmsg *txlist[100]; /* queue for when mroute reentered */
static int txrp = 0; /* txlist read pointer */
static int txwp = 0; /* txlist write pointer */
#ifdef WIN32
extern char* username;
int system_loser_win;
#endif
/**************************
** Internal Functions **
** **
**************************/
int
pvm_fd_add(fd, sets)
int fd; /* the fd */
int sets; /* which sets */
{
#ifdef SANITY
if (fd < 0 || fd >= FD_SETSIZE) {
pvmlogprintf("pvm_fd_add() bad fd %d\n", fd);
return 1;
}
#endif
if (sets & 1)
FD_SET(fd, &pvmrfds);
/*
if (sets & 2)
FD_SET(fd, &pvmwfds);
if (sets & 4)
FD_SET(fd, &pvmefds);
*/
/* if this is new highest, adjust nfds */
if (fd >= pvmnfds)
pvmnfds = fd + 1;
return 0;
}
pvm_fd_delete(fd, sets)
int fd; /* the fd */
int sets; /* which sets */
{
#ifdef SANITY
if (fd < 0 || fd >= FD_SETSIZE) {
pvmlogprintf("pvm_fd_delete() bad fd %d\n", fd);
return 1;
}
#endif
if (sets & 1)
FD_CLR(fd, &pvmrfds);
/*
if (sets & 2)
FD_CLR(fd, &pvmwfds);
if (sets & 4)
FD_CLR(fd, &pvmefds);
*/
/* if this was highest, may have to adjust nfds to new highest */
if (fd + 1 == pvmnfds)
while (pvmnfds > 0) {
pvmnfds--;
if (FD_ISSET(pvmnfds, &pvmrfds)
/*
|| FD_ISSET(pvmnfds, &pvmefds)
|| FD_ISSET(pvmnfds, &pvmwfds)
*/
) {
pvmnfds++;
break;
}
}
return 0;
}
/* ttpcb_new()
*
* Create a new, blank ttpcb.
*/
struct ttpcb *
ttpcb_new()
{
struct ttpcb *pcbp;
if (pcbp = TALLOC(1, struct ttpcb, "tpcb")) {
BZERO((char*)pcbp, sizeof(struct ttpcb));
pcbp->tt_fd = -1;
pcbp->tt_rxfrag = pmsg_new(1);
BZERO((char*)pcbp->tt_rxfrag, sizeof(struct pmsg));
pcbp->tt_rxfrag->m_link = pcbp->tt_rxfrag->m_rlink = pcbp->tt_rxfrag;
}
return pcbp;
}
/* ttpcb_delete()
*
* Get rid of a ttpcb. Delete it from any list, close the socket,
* free any rx frag heap.
*/
void
ttpcb_delete(pcbp)
struct ttpcb *pcbp;
{
struct pmsg *up;
if (pcbp->tt_link) {
LISTDELETE(pcbp, tt_link, tt_rlink);
}
if (pcbp->tt_fd != -1) {
pvm_fd_delete(pcbp->tt_fd, 3);
(void)close(pcbp->tt_fd);
}
if (up = pcbp->tt_rxfrag) {
while (up->m_link != up)
umbuf_free(up->m_link);
pmsg_unref(up);
}
if (pcbp->tt_rxf)
fr_unref(pcbp->tt_rxf);
#ifndef NOUNIXDOM
if (pcbp->tt_spath)
(void)unlink(pcbp->tt_spath);
#endif
PVM_FREE(pcbp);
}
/* ttpcb_creat()
*
* Create a new task-task entry. Allocates a ttpcb, inserts it in
* ttlist.
*/
struct ttpcb *
ttpcb_creat(tid)
int tid; /* peer tid */
{
struct ttpcb *pcbp, *pcbp2;
if (pcbp = ttpcb_new()) {
pcbp->tt_tid = tid;
for (pcbp2 = ttlist->tt_link; pcbp2 != ttlist; pcbp2 = pcbp2->tt_link)
if (pcbp2->tt_tid > tid)
break;
LISTPUTBEFORE(pcbp2, pcbp, tt_link, tt_rlink);
}
return pcbp;
}
/* ttpcb_find()
*
* Find a task-task entry by tid in ttlist.
*/
struct ttpcb *
ttpcb_find(tid)
int tid; /* peer tid */
{
struct ttpcb *pcbp;
#if defined(IMA_MPP)
if ( pcbp = mpp_ttpcb_find(tid))
return pcbp;
#endif
for (pcbp = ttlist->tt_link; pcbp != ttlist; pcbp = pcbp->tt_link)
if (pcbp->tt_tid >= tid)
break;
return (pcbp->tt_tid == tid) ? pcbp : (struct ttpcb*)0;
}
/* ttpcb_dead()
*
* Mark ttpcb dead, close socket, remove it from fd sets.
*/
void
ttpcb_dead(pcbp)
struct ttpcb *pcbp;
{
struct pmsg *up;
int sbf;
pcbp->tt_state = TTDEAD;
if (pcbp->tt_fd != -1) {
pvm_fd_delete(pcbp->tt_fd, 3);
(void)close(pcbp->tt_fd);
/* notify routing socket closed */
/*
XXX problems with the route notify messages:
XXX they can be delayed / cause deadlock because gotem isn't incremented
*/
check_routedelete(pcbp);
pcbp->tt_fd = -1;
}
#ifndef NOUNIXDOM
if (pcbp->tt_spath) {
(void)unlink(pcbp->tt_spath);
pcbp->tt_spath = 0;
}
#endif
if (pcbp->tt_rxf) {
fr_unref(pcbp->tt_rxf);
pcbp->tt_rxf = 0;
}
if (up = pcbp->tt_rxfrag) {
while (up->m_link != up)
umbuf_free(up->m_link);
}
}
int
ttpcb_dump(pcbp)
struct ttpcb *pcbp;
{
pvmlogprintf("ttpcb_dump() t%x fd=%d sad=%s",
pcbp->tt_tid,
pcbp->tt_fd,
inadport_decimal(&pcbp->tt_sad));
pvmlogprintf(" osad=%s state=%d\n",
inadport_decimal(&pcbp->tt_osad), pcbp->tt_state);
return 0;
}
int
ttpcb_dumpall()
{
struct ttpcb *pcbp;
pvmlogerror("ttpcb_dumpall()\n");
ttpcb_dump(topvmd);
for (pcbp = ttlist->tt_link; pcbp != ttlist; pcbp = pcbp->tt_link)
ttpcb_dump(pcbp);
return 0;
}
/* check_routeadd()
*
* Check if a notify is posted for new route creation, and send us
* a message if so.
*/
int
check_routeadd(pcbp)
struct ttpcb *pcbp;
{
struct waitc *wp, *wp2;
struct pmsg *up;
int sbf;
wp = waitlist->wa_link;
while (wp != waitlist) {
wp2 = wp->wa_link;
if (wp->wa_kind == WT_ROUTEA) {
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataFoo));
pvm_pkint(&pcbp->tt_tid, 1, 1);
pvm_pkint(&pcbp->tt_fd, 1, 1);
sbf = pvm_setsbuf(sbf);
up = midtobuf(sbf);
up->m_ctx = wp->wa_mesg->m_ctx;
up->m_tag = wp->wa_mesg->m_tag;
mesg_input(up);
if (wp->wa_count != -1 && --wp->wa_count < 1)
wait_delete(wp);
}
wp = wp2;
}
return 0;
}
/* check_routedelete()
*
* Check if a notify is posted for route deletion, and send us
* a message if so.
*/
int
check_routedelete(pcbp)
struct ttpcb *pcbp;
{
struct waitc *wp, *wp2;
struct pmsg *up;
int tid = pcbp->tt_tid;
wp = waitlist->wa_link;
while (wp != waitlist) {
wp2 = wp->wa_link;
if (wp->wa_kind == WT_ROUTED && tid == wp->wa_on) {
up = wp->wa_mesg;
wp->wa_mesg = 0;
mesg_input(up);
wait_delete(wp);
}
wp = wp2;
}
return 0;
}
int
post_routedelete(tid, ctx, tag)
int tid;
int ctx;
int tag;
{
int sbf;
struct waitc *wp;
struct pmsg *up;
int i;
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataFoo));
pvm_pkint(&tid, 1, 1);
i = -1;
pvm_pkint(&i, 1, 1);
sbf = pvm_setsbuf(sbf);
up = midtobuf(sbf);
up->m_ctx = ctx;
up->m_tag = tag;
if (ttpcb_find(tid)) {
wp = wait_new(WT_ROUTED);
wp->wa_tid = pvmmytid;
wp->wa_on = tid;
wp->wa_mesg = up;
} else {
mesg_input(up);
}
return 0;
}
/* pvm_tc_conreq()
*
* Another task requests a connection with us.
* Reply with a TC_CONACK message.
*
* TC_CONREQ() {
* int tdprotocol // t-d protocol revision number
* string sockaddr // address of other socket
* }
*/
static int
pvm_tc_conreq(mid)
int mid;
{
int src; /* sender of request */
int rbf; /* rx buffer temp */
int sbf = 0; /* reply message mid */
int ttpro; /* protocol revision */
struct ttpcb *pcbp; /* pcb for connection */
int ackd; /* allow connection (0) */
char *addr = ""; /* socket address */
static int linger[2] = { 1, 60 }; /* XXX arbitrary time */
int i;
int ictx;
#ifdef SOCKLENISUINT
size_t oslen;
#else
int oslen;
#endif
#ifndef NOUNIXDOM
struct sockaddr_un uns;
char spath[LEN_OF_TMP_NAM];
#endif
char buf[256];
rbf = pvm_setrbuf(mid);
pvm_bufinfo(mid, (int *)0, (int *)0, &src);
pvm_upkint(&ttpro, 1, 1);
pvm_upkstr(buf);
if (pcbp = ttpcb_find(src)) {
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("pvm_tc_conreq() crossed CONREQ from t%x\n", src);
}
if (pcbp->tt_state == TTCONWAIT) {
if (buf[0] == '/') {
#ifdef NOUNIXDOM
pvmlogprintf(
"pvm_tc_conreq() CONREQ from t%x, Unix domain socket unsupported\n",
src);
#else /*NOUNIXDOM*/
BZERO((char*)&uns, sizeof(uns));
uns.sun_family = AF_UNIX;
strcpy(uns.sun_path, buf);
while ((i = connect(pcbp->tt_fd, (struct sockaddr*)&uns,
sizeof(uns))) == -1
&& errno == EINTR)
;
if (i == -1)
pvmlogperror("pvm_tc_conreq() connect");
else
pcbp->tt_state = TTOPEN;
#endif /*NOUNIXDOM*/
} else {
pcbp->tt_osad.sin_family = AF_INET;
hex_inadport(buf, &pcbp->tt_osad);
while ((i = connect(pcbp->tt_fd,
(struct sockaddr*)&pcbp->tt_osad,
sizeof(pcbp->tt_osad))) == -1
&& errno == EINTR)
;
if (i == -1) {
pvmlogperror("pvm_tc_conreq() connect");
} else {
pcbp->tt_state = TTOPEN;
#ifndef NOSOCKOPT
if (setsockopt(pcbp->tt_fd, SOL_SOCKET, SO_LINGER,
(char*)linger, sizeof(linger)) == -1)
pvmlogperror("pvm_tc_conreq() setsockopt");
#endif /*NOSOCKOPT*/
}
}
if (pcbp->tt_state == TTOPEN) {
#ifndef WIN32
if ((i = fcntl(pcbp->tt_fd, F_GETFL, 0)) == -1)
pvmlogperror("pvm_tc_conreq() fcntl");
else {
#ifdef O_NDELAY
i |= O_NDELAY;
#else
i |= FNDELAY;
#endif
(void)fcntl(pcbp->tt_fd, F_SETFL, i);
}
#endif
pvm_fd_add(pcbp->tt_fd, 1);
}
} else {
pvmlogprintf("pvm_tc_conreq() CONREQ from t%x but state=%d ?\n",
src, pcbp->tt_state);
}
} else {
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("pvm_tc_conreq() CONREQ from t%x\n", src);
}
ackd = 1;
pcbp = ttpcb_creat(src);
if (pvmrouteopt != PvmDontRoute) {
if (buf[0] == '/') {
#ifdef NOUNIXDOM
pvmlogprintf(
"pvm_tc_conreq() CONREQ from t%x, Unix domain socket unsupported\n",
src);
#else /*NOUNIXDOM*/
if ((pcbp->tt_fd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
pvmlogperror("pvm_tc_conreq() socket");
} else {
(void)TMPNAMFUN(spath);
BZERO((char*)&uns, sizeof(uns));
uns.sun_family = AF_UNIX;
spath[0] = 0;
(void)tmpnam(spath);
strcpy(uns.sun_path, spath);
if (bind(pcbp->tt_fd, (struct sockaddr*)&uns, sizeof(uns))
== -1) {
pvmlogperror("pvm_tc_conreq() bind");
} else {
if (listen(pcbp->tt_fd, 1) == -1)
pvmlogperror("pvm_tc_conreq() listen");
else {
pcbp->tt_state = TTGRNWAIT;
pvm_fd_add(pcbp->tt_fd, 1);
ackd = 0;
addr = pcbp->tt_spath = STRALLOC(spath);
/* new route socket listening, will be accepted later */
check_routeadd(pcbp);
}
}
}
#endif /*NOUNIXDOM*/
} else {
if ((pcbp->tt_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
pvmlogperror("pvm_tc_conreq() socket");
} else {
pcbp->tt_sad = pvmourinet;
oslen = sizeof(pcbp->tt_sad);
if (bind(pcbp->tt_fd, (struct sockaddr*)&pcbp->tt_sad, oslen) == -1) {
pvmlogperror("pvm_tc_conreq() bind");
} else {
if (getsockname(pcbp->tt_fd,
(struct sockaddr*)&pcbp->tt_sad, &oslen) == -1) {
pvmlogperror("pvm_tc_conreq() getsockname");
} else {
if (listen(pcbp->tt_fd, 1) == -1)
pvmlogperror("pvm_tc_conreq() listen");
else {
hex_inadport(buf, &pcbp->tt_osad);
pcbp->tt_state = TTGRNWAIT;
pvm_fd_add(pcbp->tt_fd, 1);
ackd = 0;
addr = inadport_hex(&pcbp->tt_sad);
/* new route socket listening, will be accepted later */
check_routeadd(pcbp);
}
}
}
}
}
}
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataFoo));
ttpro = TDPROTOCOL;
pvm_pkint(&ttpro, 1, 1);
pvm_pkint(&ackd, 1, 1);
pvm_pkstr(addr);
i = pvmrescode;
pvmrescode = 1;
ictx = pvm_setcontext(SYSCTX_TC);
pvm_send(src, TC_CONACK);
pvm_setcontext(ictx);
pvmrescode = i;
pvm_freebuf(pvm_setsbuf(sbf));
if (ackd)
ttpcb_delete(pcbp);
}
pvm_setrbuf(rbf);
pvm_freebuf(mid);
return 0;
}
/* pvm_tc_conack()
*
* Another task replies to our connection request.
*
* TC_CONACK() {
* int tdprotocol // t-d protocol revision number
* int ack // 0 ok, 1 denied
* string sockaddr // address of other socket
* }
*/
static int
pvm_tc_conack(mid)
int mid;
{
int src; /* sender of reply */
int rbf; /* rx buffer temp */
int ttpro; /* protocol revision */
int ackd; /* connection allowed (0) */
struct ttpcb *pcbp; /* pcb for connection */
static int linger[2] = { 1, 60 }; /* XXX arbitrary time */
int i;
#ifndef NOUNIXDOM
struct sockaddr_un uns;
#endif
char buf[256];
rbf = pvm_setrbuf(mid);
pvm_bufinfo(mid, (int *)0, (int *)0, &src);
pvm_upkint(&ttpro, 1, 1);
pvm_upkint(&ackd, 1, 1);
pvm_upkstr(buf);
if (pcbp = ttpcb_find(src)) {
if (pcbp->tt_state == TTCONWAIT) {
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("pvm_tc_conack() CONACK from t%x\n", src);
}
if (ttpro != TDPROTOCOL) {
pvmlogprintf("pvm_tc_conack() t-t protocol mismatch with t%x\n",
pcbp->tt_tid);
ackd = 1;
} else {
if (ackd != 0) {
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("pvm_tc_conack() route to t%x denied\n",
pcbp->tt_tid);
}
} else {
if (buf[0] == '/') {
#ifdef NOUNIXDOM
pvmlogprintf(
"pvm_tc_conack() CONREQ from t%x, Unix domain socket unsupported\n",
src);
ackd = 1;
#else /*NOUNIXDOM*/
BZERO((char*)&uns, sizeof(uns));
uns.sun_family = AF_UNIX;
strcpy(uns.sun_path, buf);
while ((i = connect(pcbp->tt_fd, (struct sockaddr*)&uns,
sizeof(uns))) == -1
&& errno == EINTR)
;
if (i == -1) {
pvmlogperror("pvm_tc_conack() connect");
ackd = 1;
} else {
pcbp->tt_state = TTOPEN;
#ifndef WIN32
if ((i = fcntl(pcbp->tt_fd, F_GETFL, 0)) == -1)
pvmlogperror("pvm_tc_conack() fcntl");
else {
#ifdef O_NDELAY
i |= O_NDELAY;
#else
i |= FNDELAY;
#endif
(void)fcntl(pcbp->tt_fd, F_SETFL, i);
}
#endif
pvm_fd_add(pcbp->tt_fd, 1);
}
#endif /*NOUNIXDOM*/
} else {
pcbp->tt_osad.sin_family = AF_INET;
hex_inadport(buf, &pcbp->tt_osad);
while ((i = connect(pcbp->tt_fd,
(struct sockaddr*)&pcbp->tt_osad,
sizeof(pcbp->tt_osad))) == -1
&& errno == EINTR)
;
if (i == -1) {
pvmlogperror("pvm_tc_conack() connect");
ackd = 1;
} else {
pcbp->tt_state = TTOPEN;
#ifndef NOSOCKOPT
if (setsockopt(pcbp->tt_fd, SOL_SOCKET, SO_LINGER,
(char*)linger, sizeof(linger)) == -1)
pvmlogperror("pvm_tc_conack() setsockopt");
#endif /*NOSOCKOPT*/
#ifndef WIN32
if ((i = fcntl(pcbp->tt_fd, F_GETFL, 0)) == -1)
pvmlogperror("pvm_tc_conack() fcntl");
else {
#ifdef O_NDELAY
i |= O_NDELAY;
#else
i |= FNDELAY;
#endif
(void)fcntl(pcbp->tt_fd, F_SETFL, i);
}
pvm_fd_add(pcbp->tt_fd, 1);
#endif
}
}
}
}
if (ackd != 0) {
pcbp->tt_state = TTDENY;
(void)close(pcbp->tt_fd);
pcbp->tt_fd = -1;
}
} else {
pvmlogprintf("pvm_tc_conack() CONACK from t%x but state=%d\n",
src, pcbp->tt_state);
}
} else {
pvmlogprintf("pvm_tc_conack() suprious CONACK from t%x\n", src);
}
pvm_setrbuf(rbf);
pvm_freebuf(mid);
return 0;
}
/* pvm_tc_taskexit()
*
* We are notified that another task (to which we have a direct route)
* has exited.
*/
static int
pvm_tc_taskexit(mid)
int mid;
{
int rbf; /* rx buffer temp */
int tid; /* exited task id */
struct ttpcb *pcbp;
rbf = pvm_setrbuf(mid);
pvm_upkint(&tid, 1, 1);
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("pvm_tc_taskexit() TASKEXIT for t%x\n", tid);
}
/*
* v3.3: for now, don't close fully open route as we may lose
* messages. rely on socket getting EOF
*/
if ((pcbp = ttpcb_find(tid))
&& pcbp->tt_state != TTOPEN)
ttpcb_dead(pcbp);
pvm_setrbuf(rbf);
pvm_freebuf(mid);
return 0;
}
/* mxinput()
*
* Input from a connection.
* Reassemble packets into messages and receive them.
* Returns >= 0 the number of messages added to pvmrxlist,
* or negative on error.
*/
static int
mxinput(pcbp)
struct ttpcb *pcbp;
{
int gotem = 0; /* num msgs added to pvmrxlist */
struct frag *fp; /* shadows pcbp->tt_rxf */
struct frag *fp2;
int n; /* bytes received */
int m; /* length of fragment */
struct pmsg *rxup; /* message containing this frag */
struct pmsg *frgpile, *hdfrgpile;
char *cp; /* gp */
int src;
int dst;
int ff;
static int fairprobe = 0;
#if defined(IMA_MPP)
/* Messages from Tasks and Messages from the daemon are probed
* fairprobe makes sure that neither interface gets starved.
* fairprobe is toggled when an EOM is found
*/
switch (fairprobe)
{
/* pvm_readfrom<xxxx> returns a pointer to a frag, with its
* length properly filled in. cp returns a pointer to the
* the start of the data
*/
case 0:
if ((fp = pvm_readfrompeer()) == (struct frag *) NULL)
fp = pvm_readfrompvmd();
break;
default:
if ((fp = pvm_readfrompvmd()) == (struct frag *) NULL)
fp = pvm_readfrompeer();
break;
}
if (!fp)
return gotem; /* didn't read anything */
#else
if (!pcbp->tt_rxf)
if (!(pcbp->tt_rxf = fr_new(pvmfrgsiz)))
return PvmNoMem;
fp = pcbp->tt_rxf;
/* read fragment header separately from body */
n = (fp->fr_len < TDFRAGHDR) ? 0 : pvmget32(fp->fr_dat + 8);
n += TDFRAGHDR - fp->fr_len;
if (pvmdebmask & PDMPACKET) {
pvmlogprintf("mxinput() pcb t%x fr_len=%d fr_dat=+%d n=%d\n",
pcbp->tt_tid, fp->fr_len, fp->fr_dat - fp->fr_buf, n);
}
#ifndef WIN32
n = read(pcbp->tt_fd, fp->fr_dat + fp->fr_len, n);
#else
n = win32_read_socket( pcbp->tt_fd,fp->fr_dat+fp->fr_len,n);
#endif
if (pvmdebmask & PDMPACKET) {
pvmlogprintf("mxinput() read=%d\n", n);
}
/* deal with errors and closes */
if (n == -1 &&
#ifndef WIN32
errno != EWOULDBLOCK &&
#endif
errno != EINTR)
{
if (pvmdebmask & PDMPACKET) {
pvmlogperror("mxinput() read");
pvmlogprintf("mxinput() t%x\n", pcbp->tt_tid);
}
return PvmSysErr;
}
if (!n) {
if (pvmdebmask & PDMPACKET) {
pvmlogprintf("mxinput() t%x read EOF\n", pcbp->tt_tid);
}
return -1;
}
if (n < 1)
return gotem;
if ((fp->fr_len += n) < TDFRAGHDR)
return gotem;
/* realloc buffer if frag won't fit */
m = TDFRAGHDR + pvmget32(fp->fr_dat + 8);
if (fp->fr_len == TDFRAGHDR) {
if (m > fp->fr_max - (fp->fr_dat - fp->fr_buf)) {
fp2 = fr_new(m);
BCOPY(fp->fr_dat, fp2->fr_dat, TDFRAGHDR);
fp2->fr_len = fp->fr_len;
fr_unref(fp);
fp = pcbp->tt_rxf = fp2;
if (pvmdebmask & PDMPACKET) {
pvmlogprintf("mxinput() realloc frag max=%d\n", m);
}
}
}
#endif /* defined (IMA_MPP) */
/* Now get information from the frag header to determine
* src, destination, length, etc.
*/
#if !defined(IMA_MPP)
if (fp->fr_len == m)
#else
if (fp) /* got a frag */
#endif
{
#if !defined (IMA_MPP)
pcbp->tt_rxf = 0;
#endif
cp = fp->fr_dat;
dst = pvmget32(cp);
src = pvmget32(cp + 4);
ff = pvmget8(cp + 12);
fp->fr_len -= TDFRAGHDR;
fp->fr_dat += TDFRAGHDR;
if (pvmdebmask & PDMPACKET) {
pvmlogprintf("mxinput() pkt src t%x len %d ff %d\n",
src, fp->fr_len, ff);
}
#if defined(IMA_MPP)
if (fp -> fr_rip ) /* frag was received inplace */
{
gotem++;
fr_unref(fp); /* free frag -- mppprecv knows that the msg completed */
return gotem;
}
frgpile = hdfrgpile = pvm_mpp_pmsgs();
#else
frgpile = hdfrgpile = pcbp->tt_rxfrag ;
#endif
/*
* if start of message, make new umbuf, add to frag pile
*/
if (ff & FFSOM) {
cp += TDFRAGHDR;
fp->fr_len -= MSGHDRLEN;
fp->fr_dat += MSGHDRLEN;
rxup = umbuf_new();
rxup->m_enc = pvmget32(cp);
rxup->m_tag = pvmget32(cp + 4);
rxup->m_ctx = pvmget32(cp + 8);
rxup->m_wid = pvmget32(cp + 16);
rxup->m_crc = pvmget32(cp + 20);
rxup->m_src = src;
rxup->m_dst = dst;
LISTPUTBEFORE(hdfrgpile, rxup, m_link, m_rlink);
} else {
/* locate frag's message */
for (rxup = hdfrgpile->m_link; rxup != hdfrgpile;
rxup = rxup->m_link)
if (rxup->m_src == src)
break;
}
if (rxup == hdfrgpile) { /* uh oh, no message for it */
pvmlogerror("mxinput() frag with no message\n");
fr_unref(fp);
} else {
LISTPUTBEFORE(rxup->m_frag, fp, fr_link, fr_rlink);
rxup->m_len += fp->fr_len;
/*
* if end of message, move to pvmrxlist and count it
*/
if (ff & FFEOM) {
#if defined(IMA_MPP)
fairprobe = (fairprobe ? 0 : 1);
#endif
LISTDELETE(rxup, m_link, m_rlink);
if (pvmdebmask & PDMMESSAGE) {
pvmlogprintf(
"mxinput() src t%x route t%x ctx %d tag %s len %d\n",
rxup->m_src, (pcbp != 0 ? pcbp->tt_tid: -1),
rxup->m_ctx,
pvmnametag(rxup->m_tag, (int *)0), rxup->m_len);
}
#ifdef MCHECKSUM
if (rxup->m_crc != umbuf_crc(rxup)) {
pvmlogprintf(
"mxinput() message src t%x route t%x tag %s bad checksum\n",
rxup->m_src, pcbp->tt_tid,
pvmnametag(rxup->m_tag, (int *)0));
umbuf_free(rxup);
} else {
#endif
pmsg_setenc(rxup, rxup->m_enc);
mesg_input(rxup);
gotem++;
#ifdef MCHECKSUM
}
#endif
}
}
}
return gotem;
}
/* mxfer()
*
* Move message frags between task and pvmd or other tasks.
* Returns when
* Outgoing message (if any) fully sent AND
* (Timed out (tmout) OR
* At least one message fully received) AND
* No message from pvmd partially received.
* Returns >= 0 the number of complete messages downloaded, or
* negative on error.
*/
/* define a preallocated pool of inplace headers */
#define SHORT_PAYLOAD 128
#define HEADER_SIZE (TDFRAGHDR + MSGHDRLEN)
#define NINPLACE_HDRS (8096/(SHORT_PAYLOAD + HEADER_SIZE))
static char inPlaceHdrPool[NINPLACE_HDRS * ( HEADER_SIZE )];
static int hdrPoolStk = 0;
static int
mxfer(txup, tmout)
struct pmsg *txup; /* outgoing message or null */
struct timeval *tmout; /* time limit to get at least one message */
{
struct ttpcb *txpcbp = 0; /* route for ogm */
struct frag *txfp = 0; /* cur ogm frag or null */
struct timeval tin;
struct timeval tnow;
struct timeval *tvp;
fd_set rfds, wfds;
int wantmore = 1; /* nothing rxd yet and not timed out */
int gotem = 0; /* count of received msgs downloaded */
int ff;
int n;
#ifdef SOCKLENISUINT
size_t oslen;
#else
int oslen;
#endif
char *txcp = 0; /* point to remainder of txfp */
int txtogo = 0; /* len of remainder of txfp */
struct sockaddr_in sad;
int s;
struct ttpcb *pcbp;
char *inPlaceHeader = (char *) NULL; /* for inplace data */
int inPlaceBodyLen = 0; /* len of inplace body */
int bypassRead = FALSE; /* bypass reading packets before send?*/
int waitForOgmToComplete = FALSE;
int probedForIncomingPkts = FALSE; /* need to probe for incoming pkts */
struct msgid *sendmsg = (struct msgid *) NULL;
char errtxt[64];
if (tmout)
pvmgetclock(&tin);
if (txup) { /* transmitting a umsg */
txfp = txup->m_frag->fr_link;
if (!txfp->fr_buf) {
if (!(txfp = fr_new(MAXHDR)))
return PvmNoMem;
txfp->fr_dat += MAXHDR;
LISTPUTBEFORE(txup->m_frag, txfp, fr_link, fr_rlink);
}
/* Get connection info for this dest. If none, go to pvmd */
if (!(txpcbp = ttpcb_find(txup->m_dst)) || txpcbp->tt_state != TTOPEN)
txpcbp = topvmd;
txup->m_ref++;
bypassRead = TRUE; /* bypass reading the first time through */
}
if (pvmdebmask & PDMMESSAGE) {
pvmlogprintf("mxfer() mid %d", (txup ? txup->m_mid : 0));
if (txup)
pvmlogprintf(" ctx %d tag %s dst t%x route t%x",
txup->m_ctx,
pvmnametag(txup->m_tag, (int *)0),
txup->m_dst, txpcbp->tt_tid);
if (tmout)
pvmlogprintf(" tmout %d.%06d\n",
tmout->tv_sec, tmout->tv_usec);
else
pvmlogprintf(" tmout inf\n");
}
/* XXX we go an extra time through select because wantmore gets set
XXX inside loop. no big deal. also, could avoid calling gettimeofday
XXX if tmout is 0.0 */
/* Do this loop if
* 1) Sending a frag
* OR 2) Received *part* of an incoming frag from the daemon
* OR 3) Wantmore is true
* OR 4) Need the current OGM to complete its asynch send
* OR 5) We haven't probed at least once for incoming packets
*
* wantmore is initialized to TRUE -> loop exec'ed at least once
*/
while (txfp || topvmd->tt_rxf || wantmore || waitForOgmToComplete
|| !probedForIncomingPkts
)
{
/* Either we are sending something, or have a partial recv'd frag */
if (txfp || topvmd->tt_rxf) /* gotta block */
{
/* must block if sending any message, or receiving from pvmd */
tvp = (struct timeval *) NULL;
}
else
{
if (wantmore)
{
/* Calculate how long to wait inside the "select"
* tval = NULL means wait forever
* tval = (0,0) means probe
* tval = (x,y) means wait at most x sec + y usec
*/
if (tmout) /* finite time to block */
{
pvmgetclock(&tnow);
TVXSUBY(&tnow, &tnow, &tin);
if (TVXLTY(tmout, &tnow)) /* time's up */
{
TVCLEAR(&tnow);
wantmore = 0;
}
else
{
TVXSUBY(&tnow, tmout, &tnow);
}
tvp = &tnow;
}
else
{ /* forever to block */
tvp = (struct timeval *) NULL;
}
}
else
{ /* return asap */
TVCLEAR(&tnow);
tvp = &tnow;
}
}
/* Here we are setting the read and write file descriptors for
* the following select statement.
*/
#if !defined(IMA_MPP)
rfds = pvmrfds;
FD_ZERO(&wfds);
if (txfp)
FD_SET(txpcbp->tt_fd, &wfds);
#endif
if (pvmdebmask & PDMSELECT) {
if (tvp)
pvmlogprintf("mxfer() select timeout %d.%06d\n",
tvp->tv_sec, tvp->tv_usec);
else
pvmlogprintf("mxfer() select timeout inf\n");
#if !defined(IMA_MPP)
print_fdset("mxfer() rfds=", pvmnfds, &rfds);
print_fdset("mxfer() wfds=", pvmnfds, &wfds);
#endif
}
#if !defined(IMA_MPP)
probedForIncomingPkts = TRUE;
if ((n = select(pvmnfds,
#ifdef FDSETISINT
(int *)&rfds, (int *)&wfds, (int *)0,
#else
&rfds, &wfds, (fd_set*)0,
#endif
tvp)) == -1
&& errno != EINTR) {
pvmlogperror("mxfer() select");
return PvmSysErr;
}
if (pvmdebmask & PDMSELECT) {
pvmlogprintf("mxfer() select returns %d\n", n);
}
if (n == -1)
continue;
/*
* if pvmd conn has data ready, read packets
*/
if (FD_ISSET(topvmd->tt_fd, &rfds)
&& !(mxfersingle && gotem))
{
if ((n = mxinput(topvmd)) < 0) {
if (n != -1)
pvmlogerror("mxfer() mxinput bad return on pvmd sock\n");
else if (pvmdebmask & (PDMSELECT|PDMMESSAGE|PDMPACKET))
pvmlogerror("mxfer() EOF on pvmd sock\n");
return PvmSysErr;
}
if (gotem += n)
wantmore = 0;
}
/*
* if task conn has data ready, read packets
*/
/* This goes through *all* of the pcbs that have been defined to
* check for messages. This is too much overhead for IMA_MPP
* machines. Instead all local routes must be probed at one time.
*/
for (pcbp = ttlist->tt_link; pcbp != ttlist; pcbp = pcbp->tt_link) {
if (pcbp->tt_state == TTOPEN && FD_ISSET(pcbp->tt_fd, &rfds)) {
if ((n = mxinput(pcbp)) < 0)
ttpcb_dead(pcbp);
else {
if (gotem += n)
wantmore = 0;
}
} else /* Try and complete a direct (socket) Connection rqst */
if (pcbp->tt_state == TTGRNWAIT
&& FD_ISSET(pcbp->tt_fd, &rfds)) {
oslen = sizeof(sad);
s = accept(pcbp->tt_fd, (struct sockaddr*)&sad, &oslen);
if (s == -1) {
pvmlogperror("mxfer() accept");
ttpcb_dead(pcbp);
} else {
#ifndef NOSOCKOPT
#ifndef NOUNIXDOM
if (!pcbp->tt_spath) { /* not unix sock */
#endif
n = 1;
if (setsockopt(s, IPPROTO_TCP, TCP_NODELAY,
(char*)&n, sizeof(int)) == -1)
pvmlogperror("mxfer() setsockopt");
n = TTSOCKBUF;
if (setsockopt(s, SOL_SOCKET, SO_SNDBUF,
(char*)&n, sizeof(int)) == -1
|| setsockopt(s, SOL_SOCKET, SO_RCVBUF,
(char*)&n, sizeof(n)) == -1)
pvmlogperror("mxfer() setsockopt SO_SNDBUF");
#ifndef NOUNIXDOM
}
#endif
#endif /*NOSOCKOPT*/
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("mxfer() accept from t%x = %d\n",
pcbp->tt_tid, s);
}
(void)close(pcbp->tt_fd);
/*
(void)dup2(s, pcbp->tt_fd);
(void)close(s);
*/
pvm_fd_delete(pcbp->tt_fd, 1);
/* notify route socket closed */
check_routedelete(pcbp);
pcbp->tt_fd = s;
pcbp->tt_state = TTOPEN;
pvm_fd_add(s, 1);
/* new route socket created for communication */
check_routeadd(pcbp);
}
}
}
#else /* !IMA_MPP */
/* For Native Messaging mxinput is used to probe pvmd AND tasks when
* given a NULL ttpcb pointer
*/
if (!bypassRead)
{
probedForIncomingPkts = TRUE;
if ((n = mxinput((struct ttpcb *) NULL)) < 0) {
pvmlogerror("mxfer() mxinput bad return on probe\n");
return PvmSysErr;
}
if (gotem += n)
wantmore = 0;
}
#endif /* !IMA_MPP */
/*
* if sending and socket ready to write, send packets
*/
#if !defined(IMA_MPP)
/* XXX Native message direct routes are alway open, if they exist */
if (txfp && txpcbp->tt_state != TTOPEN)
{
/* make sure our route hasn't just been closed */
txpcbp = topvmd;
txcp = 0;
txtogo = 0;
txfp = txup->m_frag->fr_link;
}
else
#endif
{
if (txfp
&& !waitForOgmToComplete
#if !defined(IMA_MPP)
&& FD_ISSET(txpcbp->tt_fd, &wfds)
#endif
){
inPlaceHeader = (char *) NULL; /*sending inplace header ?*/
inPlaceBodyLen = 0; /* length of inplace body, if inplace*/
if (!txtogo) /* We haven't sent anything, yet, Build Header */
{
if (txfp->fr_u.dab) /* packed data */
{
txcp = txfp->fr_dat;
}
else /* inplace data */
{
txcp = inPlaceHeader = inPlaceHdrPool +
hdrPoolStk*(HEADER_SIZE+SHORT_PAYLOAD) + HEADER_SIZE;
if (pvmdebmask & PDMPACKET)
{
sprintf(errtxt,"mxfer() hdrPoolStk is %d\n", hdrPoolStk);
pvmlogerror(errtxt);
}
if (hdrPoolStk++ >= NINPLACE_HDRS)
{
sprintf(errtxt, "mxfer(): > %d inplace frags \n",
NINPLACE_HDRS);
pvmlogerror(errtxt);
return PvmSysErr;
}
}
txtogo = txfp->fr_len;
/*
* if this is first frag, prepend message header
*/
ff = 0;
if (txfp->fr_rlink == txup->m_frag)
{
txcp -= MSGHDRLEN;
txtogo += MSGHDRLEN;
pvmput32(txcp,
(txup->m_enc == 0x20000000 ? pvmmydsig : txup->m_enc));
pvmput32(txcp + 4, txup->m_tag);
pvmput32(txcp + 8, txup->m_ctx);
pvmput32(txcp + 16, txup->m_wid);
#ifdef MCHECKSUM
pvmput32(txcp + 20, umbuf_crc(txup));
#else
pvmput32(txcp + 20, 0);
#endif
ff = FFSOM;
}
if (txfp->fr_link == txup->m_frag)
ff |= FFEOM;
/*
* prepend frag header
*/
txcp -= TDFRAGHDR;
pvmput32(txcp, txup->m_dst);
pvmput32(txcp + 4, pvmmytid);
pvmput32(txcp + 8, txtogo);
pvmput32(txcp + 12, 0); /* to keep putrify happy */
pvmput8(txcp + 12, ff);
txtogo += TDFRAGHDR;
if (!txfp->fr_u.dab) /* inplace data */
{
txtogo -= txfp->fr_len;
inPlaceBodyLen = txfp->fr_len;
}
}
if (pvmdebmask & PDMPACKET)
{
pvmlogprintf("mxfer() dst t%x n=%d\n", txup->m_dst, txtogo);
}
#if defined(IMA_MPP)
/* node_send will do an async send. Feed it a midlist struct.
* when we get to the end of a message we call ogm_done
* to make sure message is really gone, and we can free
* memory
*
*/
n = pvm_node_send(txcp, txtogo, txpcbp, &sendmsg,
inPlaceHeader, inPlaceBodyLen);
#else
#if defined(IMA_RS6K) || defined(IMA_SP2MPI)
n = write(txpcbp->tt_fd, txcp, min(txtogo, 4096));
#else
#ifndef WIN32
n = write(txpcbp->tt_fd, txcp, txtogo);
#else
n = win32_write_socket(txpcbp->tt_fd,txcp,txtogo);
#endif
#endif
#endif
if (pvmdebmask & PDMPACKET) {
pvmlogprintf("mxfer() wrote %d\n", n);
}
if (n == -1 &&
#ifndef WIN32
errno != EWOULDBLOCK &&
#endif
errno != EINTR)
{
if (txpcbp == topvmd)
{
pvmlogperror("mxfer() write pvmd sock");
return PvmSysErr;
}
else
{
pvmlogperror("mxfer() write tt sock");
/* reset message and route to pvmd */
ttpcb_dead(txpcbp);
txpcbp = topvmd;
txcp = 0;
txtogo = 0;
txfp = txup->m_frag->fr_link;
}
}
if (n > 0)
{
txcp += n;
if ((txtogo -= n) <= 0)
{
if (txcp == inPlaceHeader && txfp->fr_len )
/* inplace, body length > 0 */
{
txcp = txfp->fr_dat;
txtogo = txfp->fr_len;
bypassRead = TRUE; /* try to send body without
doing another read */
}
else /* packed */
{
txcp = 0;
txfp = txfp->fr_link;
if (!txfp->fr_buf) {
txfp = 0;
waitForOgmToComplete = TRUE;
bypassRead = FALSE;
}
}
}
}
else /* we couldn't send, might be a head-head send so ... */
{
bypassRead = FALSE; /* enable reading if couldn't send */
}
}
}
/* We are waiting for an outgoing message to complete. We
can't send the next message until it does complete */
if (waitForOgmToComplete)
{
if (ogm_complete(&sendmsg))
{
pmsg_unref(txup); /* unref the message */
txup = 0;
waitForOgmToComplete = FALSE;
hdrPoolStk = 0; /* recover all the inplace headers */
}
}
/* if not sending, and not waiting for ogm to complete,
flush anything waiting in the reentered-send queue */
if (!txfp && txrp != txwp && !waitForOgmToComplete) {
txup = txlist[txrp];
if (++txrp >= sizeof(txlist)/sizeof(txlist[0]))
txrp = 0;
txfp = txup->m_frag->fr_link;
txfp = txfp->fr_buf ? txfp : 0;
if (!(txpcbp = ttpcb_find(txup->m_dst))
|| txpcbp->tt_state != TTOPEN)
txpcbp = topvmd;
}
if (pvmdebmask & PDMMESSAGE) {
pvmlogprintf("mxfer() txfp %d gotem %d tt_rxf %d\n",
(txfp ? 1 : 0), gotem, (topvmd->tt_rxf ? 1 : 0));
}
}
return gotem;
}
/* mroute()
*
* Route a message to a destination / Receive incoming messages.
* Returns when
* Outgoing message (if any) fully sent AND
* (Timed out waiting (tmout) OR
* At least one message fully received)
* Returns >= 0 the number of complete messages downloaded
* (although some may have been consumed by message-handlers),
* or negative on error.
*
* If mroute() is reentered (mroute -> mxfer -> handler -> pvm_send),
* it queues the message and returns immediately.
* The enclosing mxfer() later sends any messages in this queue.
*/
int
mroute(mid, dtid, tag, tmout)
int mid; /* message (or 0) */
int dtid; /* dest */
int tag; /* type tag */
struct timeval *tmout; /* get at least one message */
{
static int alreadyhere = 0; /* have i called myself? */
struct ttpcb *pcbp; /* pcb for foreign task */
int s; /* socket */
int sbf; /* temp TC_CONREQ message */
struct pmsg *up, *up2;
struct sockaddr_in sad;
int l;
#ifdef SOCKLENISUINT
size_t oslen;
#else
int oslen;
#endif
int cc = 0;
int gotem = 0; /* count of received messages */
static struct timeval ztv = { 0, 0 };
char *addr = 0;
#ifndef NOUNIXDOM
struct sockaddr_un uns;
char spath[LEN_OF_TMP_NAM];
#endif
if (up = midtobuf(mid)) {
up->m_dst = dtid;
up->m_tag = tag;
}
if (alreadyhere) {
if (up) {
up->m_ref++;
txlist[txwp] = up;
if (++txwp >= sizeof(txlist)/sizeof(txlist[0]))
txwp = 0;
if (txwp == txrp)
abort();
}
/* we can't wait around if reentered */
if (!tmout || tmout->tv_sec || tmout->tv_usec)
return PvmNotImpl;
return 0;
}
alreadyhere = 1;
/* if sending a message, decide whether we need to route it. */
/* Direct Routing is being set up in this section of code
* For workstations (not MPP, SHMEM)
* either Unix domain or TCP sockets are used.
* task exit notifies are set up for route closure
*
* For MPPs and SHMEM,
* connections to local tasks are always direct and require no
* handshake negotiation.
*/
if (up) { /* We've got something to send */
#if !defined(IMA_MPP) /* This is the workstation direct route setup */
/* Checks made (Workstation):
* o route option is direct
* o destination is given
* o destination is a task
* o destination is not me
* o could not find task process control block
*/
if (pvmrouteopt == PvmRouteDirect
&& dtid
&& TIDISTASK(dtid)
&& dtid != pvmmytid
&& !ttpcb_find(dtid))
{
#if !defined(NOUNIXDOM)
/* Create a UNIX domain socket to build the direct route */
if ((pvmmytid & TIDHOST) == (dtid & TIDHOST)) {
if ((s = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
pvmlogperror("mroute() socket");
} else {
(void)TMPNAMFUN(spath);
BZERO((char*)&uns, sizeof(uns));
uns.sun_family = AF_UNIX;
spath[0] = 0;
(void)tmpnam(spath);
strcpy(uns.sun_path, spath);
if (bind(s, (struct sockaddr*)&uns, sizeof(uns)) == -1) {
pvmlogperror("mroute() bind");
(void)close(s);
s = -1;
} else {
pcbp = ttpcb_creat(dtid);
pcbp->tt_fd = s;
addr = pcbp->tt_spath = STRALLOC(spath);
pcbp->tt_state = TTCONWAIT;
}
}
} else {
#endif /* !NOUNIXDOM */
/* Open a TCP direct Connection here, instead */
if ((s = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
pvmlogperror("mroute() socket");
} else {
sad = pvmourinet;
oslen = sizeof(sad);
if (bind(s, (struct sockaddr*)&sad, sizeof(sad)) == -1) {
pvmlogperror("mroute() bind");
(void)close(s);
s = -1;
} else {
if (getsockname(s, (struct sockaddr*)&sad, &oslen) == -1) {
pvmlogperror("mroute() getsockname");
(void)close(s);
s = -1;
} else {
#ifndef NOSOCKOPT
l = 1;
if (setsockopt(s, IPPROTO_TCP, TCP_NODELAY,
(char*)&l, sizeof(int)) == -1)
pvmlogperror("mroute() setsockopt");
l = TTSOCKBUF;
if (setsockopt(s, SOL_SOCKET, SO_SNDBUF,
(char*)&l, sizeof(l)) == -1
|| setsockopt(s, SOL_SOCKET, SO_RCVBUF,
(char*)&l, sizeof(l)) == -1)
pvmlogperror("mroute() setsockopt SO_SNDBUF");
#endif
pcbp = ttpcb_creat(dtid);
pcbp->tt_fd = s;
pcbp->tt_sad = sad;
pcbp->tt_state = TTCONWAIT;
addr = inadport_hex(&sad);
}
}
}
#if !defined(NOUNIXDOM)
}
#endif
/* 1. Request direct connection (TC_CONREQ)
* 2. Request Notify so that we can close dead routes
*/
if (s >= 0) { /* Socket was opened correctly */
/* Send Connection Request */
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataFoo));
l = TDPROTOCOL;
pvm_pkint(&l, 1, 1);
pvm_pkstr(addr);
sbf = pvm_setsbuf(sbf);
up2 = midtobuf(sbf);
up2->m_dst = dtid;
up2->m_tag = TC_CONREQ;
up2->m_ctx = SYSCTX_TC;
cc = mxfer(up2, &ztv);
pvm_freebuf(sbf);
/* now send a task exit notify to my pvmd */
if (cc >= 0) {
gotem += cc;
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataFoo));
l = PvmTaskExit;
pvm_pkint(&l, 1, 1);
l = SYSCTX_TC;
pvm_pkint(&l, 1, 1);
l = TC_TASKEXIT;
pvm_pkint(&l, 1, 1);
l = 1;
pvm_pkint(&l, 1, 1);
pvm_pkint(&dtid, 1, 1);
sbf = pvm_setsbuf(sbf);
up2 = midtobuf(sbf);
up2->m_dst = TIDPVMD;
up2->m_tag = TM_NOTIFY;
up2->m_ctx = SYSCTX_TM;
cc = mxfer(up2, &ztv);
pvm_freebuf(sbf);
}
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("mroute() CONREQ to t%x\n", dtid);
}
/* wait until the route pcb changes state */
while (pcbp->tt_state == TTCONWAIT) {
if ((cc = mxfer((struct pmsg *)0, (struct timeval *)0)) < 0)
break;
gotem += cc;
}
if (pcbp->tt_state == TTOPEN)
check_routeadd(pcbp);
}
}
#endif /* IMA_MPP */
}
/* now send the original message with whatever route we have */
if (cc >= 0)
if ((cc = mxfer(up, tmout)) > 0)
gotem += cc;
/* clean up any stale route pcbs */
#if !defined(IMA_MPP)
pcbp = ttlist->tt_link;
while (pcbp != ttlist)
if (pcbp->tt_state == TTDEAD) {
if (pvmdebmask & PDMROUTE) {
pvmlogprintf("mroute() freeing pcb t%x\n", pcbp->tt_tid);
}
pcbp = pcbp->tt_link;
ttpcb_delete(pcbp->tt_rlink);
} else
pcbp = pcbp->tt_link;
#endif /* IMA_MPP */
alreadyhere = 0;
return (cc < 0 ? cc : gotem);
}
/* msendrecv()
*
* Single op to send a system message (usually to our pvmd) and get
* the reply.
* Returns message handle or negative if error.
*/
int
msendrecv(other, tag, context)
int other; /* dst, src tid */
int tag; /* message tag */
int context; /* context to use for the message */
{
static int nextwaitid = 1;
int cc;
struct pmsg *up;
if (!pvmsbuf)
return PvmNoBuf;
/* send message to other */
if (pvmdebmask & PDMMESSAGE) {
pvmlogprintf("msendrecv() to t%x tag %s\n", other,
pvmnametag(tag, (int *)0));
}
pvmsbuf->m_wid = nextwaitid++;
pvmsbuf->m_ctx = context;
/*
pvmlogprintf("msendrecv() ctx=%d\n", pvmsbuf->m_ctx);
*/
if ((cc = mroute(pvmsbuf->m_mid, other, tag, (struct timeval *)0)) < 0)
return cc;
/* recv matching message from other */
for (up = pvmrxlist->m_link; 1; up = up->m_link) {
if (up == pvmrxlist) {
up = up->m_rlink;
if ((cc = mroute(0, 0, 0, (struct timeval *)0)) < 0)
return cc;
up = up->m_link;
if (up == pvmrxlist)
continue;
}
if (pvmdebmask & PDMMESSAGE) {
pvmlogprintf("msendrecv() from t%x tag %s\n",
up->m_src, pvmnametag(up->m_tag, (int *)0));
}
if (up->m_src == other && up->m_tag == tag) {
if (up->m_ctx != pvmsbuf->m_ctx) {
pvmlogprintf("msendrecv() tag %s, context doesn't match\n",
pvmnametag(up->m_tag, (int *)0));
}
if (up->m_wid != pvmsbuf->m_wid) {
pvmlogprintf("msendrecv() tag %s, waitid doesn't match\n",
pvmnametag(up->m_tag, (int *)0));
}
break;
}
}
LISTDELETE(up, m_link, m_rlink);
if (pvmrbuf)
umbuf_free(pvmrbuf);
pvmrbuf = 0;
if (cc = pvm_setrbuf(up->m_mid))
return cc;
return up->m_mid;
}
static int
int_compare(i, j)
#if defined(IMA_SGI) || defined(IMA_SGI5) || defined(IMA_SGI64) || defined(IMA_SGIMP) || defined(IMA_SGIMP64)
const void *i, *j;
#else
void *i, *j;
#endif
{
return *((int *)i) - *((int *)j);
}
/* pvmmcast()
*
* Multicast a message to a list of destinations.
*/
int
pvmmcast(mid, tids, count, tag)
int mid;
int *tids;
int count;
int tag;
{
static struct timeval ztv = { 0, 0 };
int *dst;
int i, j;
int cc = 0;
struct ttpcb *pcbp;
int sbf;
/*
* make sorted list of destinations
*/
dst = TALLOC(count, int, "mcal");
BCOPY(tids, dst, count * sizeof(int));
qsort((char*)dst, count, sizeof(int), int_compare);
/*
* remove duplicates
*/
j = 0;
for (i = 1; i < count; i++)
if (dst[i] != dst[j])
dst[++j] = dst[i];
count = j + 1;
/* Need to set the context of the message buffer */
pvmsbuf->m_ctx = pvmmyctx;
/*
* remove self from list
* send over any direct routes we have
*
* XXX we should attempt new routes here if RouteDirect is on.
*/
j = 0;
for (i = 0; i < count; i++) {
if (dst[i] == pvmmytid)
continue;
if ((pcbp = ttpcb_find(dst[i])) && pcbp->tt_state == TTOPEN)
mroute(pvmsbuf->m_mid, dst[i], tag, &ztv);
else
dst[j++] = dst[i];
}
count = j; /* destinations still to go */
/*
* send rest of addresses to pvmd for distribution
*/
if (count > 0) {
/* announce multicast address list to pvmd */
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataFoo));
pvm_pkint(&count, 1, 1);
pvm_pkint(dst, count, 1);
sbf = pvm_setsbuf(sbf);
if ((cc = mroute(sbf, TIDPVMD, TM_MCA, &ztv)) > 0)
cc = 0;
pvm_freebuf(sbf);
/* send message */
if (cc >= 0)
if ((cc = mroute(pvmsbuf->m_mid, pvmmytid | TIDGID, tag, &ztv)) > 0)
cc = 0;
}
PVM_FREE(dst);
return cc;
}
/* mksocs()
*
* Make socket for talking to pvmd. Connect to address advertized
* in pvmd socket-name file.
*
* Returns 0 if okay, else error code.
*/
static int
mksocs()
{
char buf[128];
#ifndef WIN32
int d;
#else
HANDLE d;
#endif
#ifdef SOCKLENISUINT
size_t oslen;
#else
int oslen;
#endif
int n;
int try;
char *p;
#ifndef NOUNIXDOM
struct sockaddr_un uns;
#endif
if (topvmd)
return 0;
/*
* get addr of pvmd, make socket to talk.
* first try envar PVMSOCK, then try sockaddr file.
*/
if (!(p = getenv("PVMSOCK"))) {
if (!(p = pvmdsockfile())) {
pvmlogerror("mksocs() pvmdsockfile() failed\n");
goto bail;
}
#ifndef WIN32
if ((d = open(p, O_RDONLY, 0)) == -1) {
#else
d = win32_open_file(p);
if (d == (HANDLE) -2) {
system_loser_win=TRUE;
d = _open(p,O_RDONLY,0);
}
if (d== (HANDLE) -1) {
#endif
pvmlogperror(p);
goto bail;
}
#ifdef WIN32
if (!system_loser_win) {
n = win32_read_file(d,buf,sizeof(buf));
win32_close_file(d);
}
else {
n = _read(d,buf,sizeof(buf));
(void)_close(d);
}
#else
n = read(d, buf, sizeof(buf));
(void)close(d);
#endif
if (n == -1) {
pvmlogperror("mksocs() read addr file");
goto bail;
}
if (n == 0) {
pvmlogerror("mksocs() read addr file: wrong length read\n");
goto bail;
}
buf[n] = 0;
p = buf;
}
FD_ZERO(&pvmrfds);
/*
FD_ZERO(&pvmwfds);
*/
pvmnfds = 0;
topvmd = ttpcb_new();
topvmd->tt_tid = TIDPVMD;
if (p[0] == '/') {
#ifdef NOUNIXDOM
pvmlogerror("mksocs() no support for Unix domain socket\n");
goto bail;
#else /*NOUNIXDOM*/
if ((topvmd->tt_fd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) {
pvmlogperror("mksocs() socket");
goto bail;
}
/*
* XXX these 2 loops are a hack -
* XXX it keeps too many tasks from overflowing the pvmd listen q.
* XXX maybe the pvmd should make a socket pair for the task instead.
*/
try = 5;
while (1) {
BZERO((char*)&uns, sizeof(uns));
uns.sun_family = AF_UNIX;
strcpy(uns.sun_path, p);
n = sizeof(uns);
if (connect(topvmd->tt_fd, (struct sockaddr*)&uns, n) == -1) {
if (--try <= 0) {
pvmlogperror("mksocs() connect");
goto bail;
}
sleep(1); /* XXX hmm again */
} else
break;
}
#endif /*NOUNIXDOM*/
} else {
if ((topvmd->tt_fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
pvmlogperror("mksocs() socket");
goto bail;
}
try = 5;
while (1) { /* goto is at the root of all good programming style */
hex_inadport(p, &topvmd->tt_osad);
topvmd->tt_osad.sin_family = AF_INET;
n = sizeof(topvmd->tt_osad);
if (connect(topvmd->tt_fd, (struct sockaddr*)&topvmd->tt_osad, n)
== -1) {
if (--try <= 0) {
pvmlogperror("mksocs() connect");
goto bail;
}
#ifndef WIN32
sleep(1); /* XXX hmm again */
#else
Sleep(1);
#endif
} else
break;
}
#ifndef NOSOCKOPT
d = 1;
if (setsockopt(topvmd->tt_fd, IPPROTO_TCP, TCP_NODELAY, (char*)&d, sizeof(int))
== -1) {
pvmlogperror("mksocs() setsockopt");
goto bail;
}
#endif
oslen = sizeof(topvmd->tt_sad);
if (getsockname(topvmd->tt_fd, (struct sockaddr*)&topvmd->tt_sad, &oslen) == -1) {
pvmlogperror("mksocs() getsockname");
goto bail;
}
}
topvmd->tt_state = TTOPEN;
pvm_fd_add(topvmd->tt_fd, 1);
return 0;
bail:
if (topvmd)
ttpcb_delete(topvmd);
topvmd = 0;
return PvmSysErr;
}
/* unmksocs()
*
* Close socket to local pvmd.
*/
static int
unmksocs()
{
if (!topvmd)
return 1;
while (ttlist->tt_link != ttlist)
ttpcb_delete(ttlist->tt_link);
ttpcb_delete(topvmd);
topvmd = 0;
return 0;
}
/* pvmbeatask()
*
* Initialize libpvm, config process as a task.
* This is called as the first step of each libpvm function so no
* explicit initialization is required.
*
* Makes socket to talk to local pvmd, connects to pvmd, trades
* config information and authentication.
*
* Returns 0 if okay, else error code.
*/
int
pvmbeatask()
{
int sbf = 0, rbf = 0; /* saved rx and tx message handles */
int prver; /* protocol version */
int cookie; /* cookie assigned by pvmd for ident */
int cc;
char authfn[LEN_OF_TMP_NAM]; /* auth file name */
int authfd = -1; /* auth fd to validate pvmd ident */
int i;
char buf[16]; /* for converting sockaddr */
char *p;
struct pvmminfo minfo;
int outtid, outctx, outtag;
int trctid, trcctx, trctag;
int need_trcinfo = 0;
int new_tracer = 0;
char tmask[ 2 * TEV_MASK_LENGTH ];
int tbuf, topt;
int mid;
TEV_DECLS
/*
pvm_setopt(PvmDebugMask, -1);
*/
if (pvmmytid != -1)
return 0;
pvmmydsig = pvmgetdsig();
TEV_EXCLUSIVE;
authfn[0] = 0;
#ifndef WIN32
pvmmyupid = getpid();
#else
pvmmyupid = _getpid(); /* .... */
#endif
/*
* get expected pid from environment in case we were started by
* the pvmd and someone forked again
*/
if (p = getenv("PVMEPID"))
cookie = atoi(p);
else
cookie = 0;
/* if ((pvm_useruid = getuid()) == -1) */
#ifndef WIN32
if ((pvm_useruid = geteuid()) == -1) {
pvmlogerror("can't getuid()\n");
cc = PvmSysErr;
goto bail2;
}
#else
if (!username)
username = MyGetUserName();
#endif
if (p = getenv("PVMTASKDEBUG")) {
pvmdebmask = pvmstrtoi(p);
if (pvmdebmask) {
pvmlogprintf("task debug mask is 0x%x (%s)\n",
pvmdebmask, debug_flags(pvmdebmask));
}
}
#ifndef IMA_MPP
if (cc = mksocs()) /* make socket to talk to pvmd */
goto bail2;
#endif
/*
* initialize received-message list
*/
pvmrxlist = pmsg_new(1);
BZERO((char*)pvmrxlist, sizeof(struct pmsg));
pvmrxlist->m_link = pvmrxlist->m_rlink = pvmrxlist;
ttlist = TALLOC(1, struct ttpcb, "tpcb");
BZERO((char*)ttlist, sizeof(struct ttpcb));
ttlist->tt_link = ttlist->tt_rlink = ttlist;
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataFoo));
rbf = pvm_setrbuf(0);
/*
* create empty t-auth file that pvmd must write and we'll check later
*/
#ifndef IMA_MPP
(void)TMPNAMFUN(authfn);
#ifdef IMA_OS2
if ((authfd = open(authfn, O_RDWR|O_CREAT|O_TRUNC, 0600)) == -1) {
#else
if ((authfd = open(authfn, O_RDONLY|O_CREAT|O_TRUNC, 0600)) == -1) {
#endif
pvmlogperror(authfn);
pvmlogerror("pvmbeatask() can't creat t-auth file\n");
cc = PvmSysErr;
goto bail2;
}
/*
* send first connect message to pvmd
*/
mxfersingle = 1;
prver = TDPROTOCOL;
pvm_pkint(&prver, 1, 1);
pvm_pkstr(authfn);
if ((cc = msendrecv(TIDPVMD, TM_CONNECT, SYSCTX_TM)) <= 0)
goto bail;
pvm_upkint(&prver, 1, 1);
if (prver != TDPROTOCOL) {
pvmlogprintf("pvmbeatask() t-d protocol mismatch (%d/%d)\n",
TDPROTOCOL, prver);
cc = PvmSysErr;
goto bail;
}
pvm_upkint(&cc, 1, 1);
if (!cc) {
pvmlogerror("pvmbeatask() pvmd refuses connection\n");
goto bail;
}
/*
* check our t-auth file; write in pvmd d-auth file
*/
if ((cc = read(authfd, (char*)&cc, 1)) == -1) {
pvmlogperror("pvmbeatask() read authfile");
cc = PvmSysErr;
goto bail;
}
if (cc != 1) {
pvmlogerror("pvmbeatask() pvmd didn't validate itself\n");
cc = PvmSysErr;
goto bail;
}
(void)close(authfd);
(void)unlink(authfn);
pvm_upkstr(authfn);
if ((authfd = open(authfn, O_WRONLY, 0)) == -1) {
pvmlogperror(authfn);
pvmlogerror("pvmbeatask() failed to open d-auth file\n");
authfn[0] = 0;
cc = PvmSysErr;
goto bail;
}
cc = write(authfd, authfn, 1);
(void)close(authfd);
if (cc != 1) {
if (cc == -1)
pvmlogperror(authfn);
pvmlogerror("pvmbeatask() can't write d-auth file\n");
authfn[0] = 0;
cc = PvmSysErr;
goto bail;
}
authfd = -1;
authfn[0] = 0;
/*
* send second connect message to pvmd
*/
pvm_initsend(PvmDataFoo);
pvm_pkint(&pvmmyupid, 1, 1);
pvm_pkint(&cookie, 1, 1);
if ((cc = msendrecv(TIDPVMD, TM_CONN2, SYSCTX_TM)) <= 0)
goto bail;
pvm_upkint(&cc, 1, 1);
if (!cc) {
pvmlogerror("pvmbeatask() pvmd refuses connection\n");
goto bail;
}
#endif /* IMA_MPP */
#ifdef IMA_MPP
pvm_mpp_beatask(&pvmmytid, &pvmmyptid, &outtid, &outctx,
&outtag, &trctid, &trcctx,&trctag, &pvmudpmtu,
&pvmschedtid, &topvmd);
#else
pvm_upkint(&pvmmytid, 1, 1);
pvm_upkint(&pvmmyptid, 1, 1);
pvm_upkint(&outtid, 1, 1);
pvm_upkint(&outctx, 1, 1);
pvm_upkint(&outtag, 1, 1);
#endif
if (!pvmtrc.outtid) {
pvmtrc.outtid = outtid;
pvmtrc.outctx = outctx;
pvmtrc.outtag = outtag;
pvmctrc.outtid = pvmtrc.outtid;
pvmctrc.outctx = pvmtrc.outctx;
pvmctrc.outtag = pvmtrc.outtag;
}
#ifndef IMA_MPP
pvm_upkint(&trctid, 1, 1);
pvm_upkint(&trcctx, 1, 1);
pvm_upkint(&trctag, 1, 1);
#endif
if (!pvmtrc.trctid) {
pvmtrc.trctid = trctid;
pvmtrc.trcctx = trcctx;
pvmtrc.trctag = trctag;
pvmctrc.trctid = pvmtrc.trctid;
pvmctrc.trcctx = pvmtrc.trcctx;
pvmctrc.trctag = pvmtrc.trctag;
new_tracer++;
}
if (p = getenv("PVMCTX"))
pvmmyctx = pvmstrtoi(p);
/*
if (pvmmyctx == 0)
pvmmyctx = pvm_newcontext();
*/
/* get trace mask from envar or zero it */
if ( (p = getenv("PVMTMASK")) ) {
if ( strlen(p) + 1 == TEV_MASK_LENGTH )
BCOPY(p, pvmtrc.tmask, TEV_MASK_LENGTH);
else
TEV_MASK_INIT(pvmtrc.tmask);
} else {
TEV_MASK_INIT(pvmtrc.tmask);
if ( new_tracer ) need_trcinfo++;
}
BCOPY(pvmtrc.tmask, pvmctrc.tmask, TEV_MASK_LENGTH);
/* get trace buffering from envar */
if ((p = getenv("PVMTRCBUF")))
pvmtrc.trcbuf = atoi( p );
else {
pvmtrc.trcbuf = 0;
if ( new_tracer ) need_trcinfo++;
}
pvmctrc.trcbuf = pvmtrc.trcbuf;
/* get trace options from envar */
if ((p = getenv("PVMTRCOPT")))
pvmtrc.trcopt = atoi( p );
else {
pvmtrc.trcopt = 0;
if ( new_tracer ) need_trcinfo++;
}
pvmctrc.trcopt = pvmtrc.trcopt;
#ifndef IMA_MPP
pvm_upkint(&pvmudpmtu, 1, 1);
#endif
pvmfrgsiz = pvmudpmtu;
#ifndef IMA_MPP
pvm_upkint(&i, 1, 1); /* XXX data signature */
pvm_upkstr(buf);
hex_inadport(buf, &pvmourinet);
pvmourinet.sin_family = AF_INET;
pvmourinet.sin_port = 0;
pvm_upkint(&pvmschedtid, 1, 1);
#endif
wait_init(pvmmytid, TIDLOCAL);
BZERO(&minfo, sizeof(minfo));
minfo.src = -1;
minfo.ctx = SYSCTX_TC;
minfo.tag = TC_CONREQ;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_conreq);
minfo.tag = TC_CONACK;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_conack);
minfo.tag = TC_TASKEXIT;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_taskexit);
minfo.tag = TC_NOOP;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_noop);
minfo.tag = TC_SETTRACE;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_settrace);
minfo.tag = TC_SETTRCBUF;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_settrcbuf);
minfo.tag = TC_SETTRCOPT;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_settrcopt);
minfo.tag = TC_SETTMASK;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_settmask);
minfo.tag = TC_SIBLINGS;
pvm_addmhf(minfo.src, minfo.tag, minfo.ctx, pvm_tc_siblings);
pvm_freebuf(pvm_setrbuf(rbf));
pvm_freebuf(pvm_setsbuf(sbf));
mxfersingle = 0;
if ( need_trcinfo )
{
rbf = pvm_setrbuf( 0 );
if ( pvm_recvinfo( PVMTRACERCLASS, 0, PvmMboxDefault ) > 0 )
{
pvm_upkint(&trctid, 1, 1);
pvm_upkint(&trcctx, 1, 1);
pvm_upkint(&trctag, 1, 1);
pvm_upkint(&outctx, 1, 1); /* unused here */
pvm_upkint(&outtag, 1, 1); /* unused here */
pvm_upkstr(tmask);
pvm_upkint(&tbuf, 1, 1);
pvm_upkint(&topt, 1, 1);
if ( pvmtrc.trctid == trctid && pvmtrc.trcctx == trcctx
&& pvmtrc.trctag == trctag )
{
if ( strlen(tmask) + 1 == TEV_MASK_LENGTH ) {
BCOPY(tmask, pvmtrc.tmask, TEV_MASK_LENGTH);
BCOPY(pvmtrc.tmask, pvmctrc.tmask, TEV_MASK_LENGTH);
}
pvmtrc.trcbuf = tbuf;
pvmctrc.trcbuf = pvmtrc.trcbuf;
pvmtrc.trcopt = topt;
pvmctrc.trcopt = pvmtrc.trcopt;
}
pvm_freebuf(pvm_setrbuf(rbf));
}
else
pvm_setrbuf(rbf);
}
tev_init();
if (TEV_AMEXCL) {
TEV_ENDEXCL;
}
return 0;
bail:
if (pvm_getrbuf() > 0)
pvm_freebuf(pvm_getrbuf());
if (pvm_getsbuf() > 0)
pvm_freebuf(pvm_getsbuf());
pvm_setrbuf(rbf);
pvm_setsbuf(sbf);
#ifndef IMA_MPP
if (authfd != -1)
(void)close(authfd);
if (authfn[0])
(void)unlink(authfn);
unmksocs();
#endif
bail2:
if (TEV_AMEXCL) {
TEV_ENDEXCL;
}
return cc;
}
int
pvmendtask()
{
if (pvmmytid != -1) {
#if !defined(IMA_MPP)
unmksocs();
#endif
pvmmytid = -1;
pvmtrc.trctid = 0;
}
return 0;
}
void
check_for_exit(src)
int src;
{
}
/************************
** Libpvm Functions **
** **
************************/
int
pvm_getfds(fds) /* XXX this function kinda sucks */
int **fds; /* fd list return */
{
static int *fdlist = 0;
static int fdlen = 0;
int cc;
int nfds;
struct ttpcb *pcbp;
TEV_DECLS
if (TEV_EXCLUSIVE) {
if (TEV_DO_TRACE(TEV_GETFDS,TEV_EVENT_ENTRY))
TEV_FIN;
}
if (!(cc = BEATASK)) {
nfds = 1;
for (pcbp = ttlist->tt_link; pcbp != ttlist; pcbp = pcbp->tt_link)
if (pcbp->tt_state == TTOPEN || pcbp->tt_state == TTGRNWAIT)
nfds++;
if (fdlen < nfds) {
fdlen = (nfds * 3) / 2 + 1;
if (fdlist)
fdlist = TREALLOC(fdlist, fdlen, int);
else
fdlist = TALLOC(fdlen, int, "fdlist");
}
fdlist[0] = topvmd->tt_fd;
nfds = 1;
for (pcbp = ttlist->tt_link; pcbp != ttlist; pcbp = pcbp->tt_link)
if (pcbp->tt_state == TTOPEN || pcbp->tt_state == TTGRNWAIT)
fdlist[nfds++] = pcbp->tt_fd;
*fds = fdlist;
cc = nfds;
}
if (TEV_AMEXCL) {
if (TEV_DO_TRACE(TEV_GETFDS,TEV_EVENT_EXIT)) {
TEV_PACK_INT( TEV_DID_CC, TEV_DATA_SCALAR, &cc, 1, 1 );
if (cc > 0) {
TEV_PACK_INT( TEV_DID_FDS, TEV_DATA_ARRAY,
(int *)fdlist, nfds, 1 );
}
TEV_FIN;
}
TEV_ENDEXCL;
}
return (cc < 0 ? lpvmerr("pvm_getfds", cc) : cc);
}
int
pvm_start_pvmd(argc, argv, block)
int argc; /* number of args to pass to pvmd (>= 0) */
char **argv; /* args for pvmd or null */
int block; /* if true, don't return until add hosts are started */
{
char *sfn;
struct stat sb;
int cc;
char *fn; /* file to exec */
char **av;
int pfd[2];
int n;
FILE *ff;
char buf[128];
#ifdef WIN32
char *pathending = "/lib/win32/pvmd3.exe";
SECURITY_ATTRIBUTES saPipe;
PROCESS_INFORMATION pi;
STARTUPINFO si; /* for CreateProcess call */
HANDLE pvmdid;
int i = 0;
char cmd[256];
#endif
TEV_DECLS;
if (TEV_EXCLUSIVE) {
if (pvmmytid != -1
&& TEV_DO_TRACE(TEV_START_PVMD,TEV_EVENT_ENTRY)) {
TEV_PACK_INT( TEV_DID_BF, TEV_DATA_SCALAR, &block, 1, 1 );
TEV_PACK_STRING( TEV_DID_AS, TEV_DATA_ARRAY,
argv, argc, 1 );
TEV_FIN;
}
}
if (argc < 0 || !argv)
argc = 0;
#ifndef WIN32
if ((pvm_useruid = getuid()) == -1) {
pvmlogerror("can't getuid()\n");
cc = PvmSysErr;
goto bail;
}
#else
if (!username)
username = MyGetUserName();
#endif
if (!(sfn = pvmdsockfile())) {
pvmlogerror("pvm_start_pvmd() pvmdsockfile() failed\n");
cc = PvmSysErr;
goto bail;
}
if (stat(sfn, &sb) != -1) {
cc = PvmDupHost;
goto bail;
}
#ifdef WIN32
cc = PvmSysErr;
fn = malloc(128 * sizeof(char));
strcpy(fn,(char *) pvmgetroot());
strcat(fn,(char *) pathending);
av = TALLOC(argc + 2, char *, "argv");
if (argc > 0)
BCOPY((char *)&argv[0], (char *)&av[1], argc * sizeof(char*));
av[0] = fn;
if (stat(av[0],&sb) == -1) {
fprintf(stderr,"Couldn't find daemon executable !\n");
goto bail;
}
av[argc + 1] = 0;
strcpy(cmd,"");
for (i=0;i <= argc ;i++) {
strcat(cmd,av[i]);
strcat(cmd," ");
}
/* pvmdid gives us back the process handle
but it is an int anyway */
saPipe.nLength = sizeof(SECURITY_ATTRIBUTES);
saPipe.lpSecurityDescriptor = NULL;
saPipe.bInheritHandle = FALSE;
memset(&si, 0, sizeof(si));
si.cb = sizeof(si);
pvmdid = (HANDLE) CreateProcess(
av[0], /* filename */
cmd, /* full command line for child */
/* for example : hostfiles */
NULL, /* process security descriptor */
NULL, /* thread security descriptor */
FALSE, /* inherit handles? */
DETACHED_PROCESS, /* creation flags */
NULL, /* inherited environment address */
NULL, /* startup dir */
/* NULL = start in current */
&si, /* pointer to startup info (input) */
&pi); /* pointer to process info (output) */
if (!pvmdid) {
fprintf(stderr,"CreateProcess() failed \n");
fprintf(stderr,"enough memory ?\n");
goto bail;
return 1;
}
CloseHandle(pi.hThread);
CloseHandle(pi.hProcess);
PVM_FREE(av);
while (stat(sfn, &sb) == -1)
Sleep(5000); /* what a hack */
#else
#ifdef IMA_TITN
if (socketpair(AF_UNIX, SOCK_STREAM, 0, pfd) == -1)
#else
if (pipe(pfd) == -1)
#endif
{
cc = PvmSysErr;
goto bail;
}
fn = pvmgetpvmd();
av = TALLOC(argc + 2, char *, "argv");
if (argc > 0)
BCOPY((char *)&argv[0], (char *)&av[1], argc * sizeof(char*));
av[0] = fn;
av[argc + 1] = 0;
if (!fork()) {
(void)close(pfd[0]);
/* fork again so the pvmd is not the child - won't have to wait() for it */
if (!fork()) {
if (pfd[1] != 1)
dup2(pfd[1], 1);
for (n = getdtablesize(); n-- > 0; )
if (n != 1)
(void)close(n);
(void)open("/dev/null", O_RDONLY, 0); /* should be 0 */
(void)open("/dev/null", O_WRONLY, 0); /* should be 2 */
(void)signal(SIGINT, SIG_IGN);
(void)signal(SIGQUIT, SIG_IGN);
#ifndef IMA_OS2
(void)signal(SIGTSTP, SIG_IGN);
#endif
#ifdef IMA_OS2
sprintf(buf,"%s/lib/%s/pvmd3.exe",
getenv("PVM_ROOT"),getenv("PVM_ARCH"));
av[0]=buf;
#endif
execvp(av[0], av);
}
_exit(0);
}
(void)close(pfd[1]);
(void)wait(0);
PVM_FREE(av);
if (!(ff = fdopen(pfd[0], "r"))) {
cc = PvmSysErr;
(void)close(pfd[0]);
goto bail;
}
strcpy(buf, "PVMSOCK=");
n = strlen(buf);
if (!fgets(buf + n, sizeof(buf) - n - 1, ff)) {
cc = PvmCantStart;
fclose(ff);
goto bail;
}
fclose(ff);
if (strlen(buf + n) < 2) {
cc = PvmCantStart;
goto bail;
}
n = strlen(buf);
if (buf[n - 1] == '\n')
buf[n - 1] = 0;
pvmputenv(STRALLOC(buf));
/* fprintf(stderr, "pvm_start_pvmd: %s\n", buf); */
#endif
if (cc = BEATASK)
goto bail;
if (block) {
struct pvmhostinfo *hip;
int t = 1;
pvm_config((int*)0, (int*)0, &hip);
while ((cc = pvm_addhosts(&hip[0].hi_name, 1, (int*)0)) == PvmAlready) {
#ifndef WIN32
sleep(t);
#else
Sleep(t);
#endif
if (t < 8)
t *= 2;
}
if (cc != PvmDupHost)
goto bail;
cc = BEATASK;
}
bail:
if (TEV_AMEXCL) {
if (TEV_DO_TRACE(TEV_START_PVMD,TEV_EVENT_EXIT)) {
TEV_PACK_INT( TEV_DID_CC, TEV_DATA_SCALAR, &cc, 1, 1 );
TEV_FIN;
}
TEV_ENDEXCL;
}
return (cc < 0 ? lpvmerr("pvm_start_pvmd", cc) : 0);
}
int
pvm_precv(tid, tag, cp, len, dt, rtid, rtag, rlen)
int tid;
int tag;
void *cp;
int len;
int dt;
int *rtid;
int *rtag;
int *rlen;
{
int nb, mc, src;
int rbf;
int cc = 0;
long ad;
TEV_DECLS
if (TEV_EXCLUSIVE) {
if (TEV_DO_TRACE(TEV_PRECV,TEV_EVENT_ENTRY)) {
TEV_PACK_INT( TEV_DID_RST, TEV_DATA_SCALAR, &tid, 1, 1 );
TEV_PACK_INT( TEV_DID_RMC, TEV_DATA_SCALAR, &tag, 1, 1 );
ad = (long)cp;
TEV_PACK_LONG( TEV_DID_PDA, TEV_DATA_SCALAR, &ad, 1, 1 );
TEV_PACK_INT( TEV_DID_PC, TEV_DATA_SCALAR, &len, 1, 1 );
TEV_PACK_INT( TEV_DID_PDT, TEV_DATA_SCALAR, &dt, 1, 1 );
TEV_FIN;
}
}
switch (dt) {
case PVM_BYTE:
len *= sizeof(char);
break;
case PVM_SHORT:
case PVM_USHORT:
len *= sizeof(short);
break;
case PVM_INT:
case PVM_UINT:
len *= sizeof(int);
break;
case PVM_LONG:
case PVM_ULONG:
len *= sizeof(long);
break;
case PVM_FLOAT:
len *= sizeof(float);
break;
case PVM_CPLX:
len *= sizeof(float) * 2;
break;
case PVM_DOUBLE:
len *= sizeof(double);
break;
case PVM_DCPLX:
len *= sizeof(double) * 2;
break;
case PVM_STR:
cc = PvmNotImpl;
break;
default:
cc = PvmBadParam;
break;
}
if (!cc) {
#if !defined(IMA_MPP)
rbf = pvm_setrbuf(0);
cc = pvm_recv(tid, tag);
if (cc > 0) {
pvm_bufinfo(cc, &nb, &mc, &src);
if (rlen)
*rlen = nb;
if (nb < len)
len = nb;
if (rtag)
*rtag = mc;
if (rtid)
*rtid = src;
pvm_upkbyte((char *)cp, len, 1);
pvm_freebuf(cc);
cc = 0;
}
pvm_setrbuf(rbf);
#else
cc = pvm_mppprecv(tid, tag, cp ,len, dt, rtid, rtag, rlen);
#endif
}
if (TEV_AMEXCL) {
if (TEV_DO_TRACE(TEV_PRECV,TEV_EVENT_EXIT)) {
TEV_PACK_INT( TEV_DID_CC, TEV_DATA_SCALAR, &cc, 1, 1 );
if ( cc < 0 )
nb = mc = src = -1;
TEV_PACK_INT( TEV_DID_MNB, TEV_DATA_SCALAR, &nb, 1, 1 );
TEV_PACK_INT( TEV_DID_MC, TEV_DATA_SCALAR, &mc, 1, 1 );
TEV_PACK_INT( TEV_DID_SRC, TEV_DATA_SCALAR, &src, 1, 1 );
TEV_FIN;
}
TEV_ENDEXCL;
}
if (cc < 0)
lpvmerr("pvm_precv", cc);
return cc;
}
int
pvm_psend(tid, tag, cp, len, dt)
int tid;
int tag;
void *cp;
int len;
int dt;
{
int sbf;
int cc = 0;
long ad;
TEV_DECLS
if (TEV_EXCLUSIVE) {
if (TEV_DO_TRACE(TEV_PSEND,TEV_EVENT_ENTRY)) {
TEV_PACK_INT( TEV_DID_DST, TEV_DATA_SCALAR, &tid, 1, 1 );
TEV_PACK_INT( TEV_DID_MC, TEV_DATA_SCALAR, &tag, 1, 1 );
ad = (long)cp;
TEV_PACK_LONG( TEV_DID_PDA, TEV_DATA_SCALAR, &ad, 1, 1 );
TEV_PACK_INT( TEV_DID_PC, TEV_DATA_SCALAR, &len, 1, 1 );
TEV_PACK_INT( TEV_DID_PDT, TEV_DATA_SCALAR, &dt, 1, 1 );
TEV_FIN;
}
}
switch (dt) {
case PVM_BYTE:
len *= sizeof(char);
break;
case PVM_SHORT:
case PVM_USHORT:
len *= sizeof(short);
break;
case PVM_INT:
case PVM_UINT:
len *= sizeof(int);
break;
case PVM_LONG:
case PVM_ULONG:
len *= sizeof(long);
break;
case PVM_FLOAT:
len *= sizeof(float);
break;
case PVM_CPLX:
len *= sizeof(float) * 2;
break;
case PVM_DOUBLE:
len *= sizeof(double);
break;
case PVM_DCPLX:
len *= sizeof(double) * 2;
break;
case PVM_STR:
cc = PvmNotImpl;
break;
default:
cc = PvmBadParam;
break;
}
#if !defined(IMA_MPP)
if (!cc) {
sbf = pvm_setsbuf(pvm_mkbuf(PvmDataInPlace));
pvm_pkbyte((char *)cp, len, 1);
if ((cc = pvm_send(tid, tag)) > 0)
cc = 0;
pvm_freebuf(pvm_setsbuf(sbf));
}
#else
cc = pvm_mpppsend((char *) cp, len, tid, tag);
#endif
if (TEV_AMEXCL) {
if (TEV_DO_TRACE(TEV_PSEND,TEV_EVENT_EXIT)) {
TEV_PACK_INT( TEV_DID_CC, TEV_DATA_SCALAR, &cc, 1, 1 );
TEV_FIN;
}
TEV_ENDEXCL;
}
if (cc < 0)
lpvmerr("pvm_psend", cc);
return cc;
}
/* ----- ogm_complete ------ */
int
ogm_complete( ogmlist )
struct msgid **ogmlist;
{
struct msgid *mp, *oldmp, *head;
#if defined(IMA_MPP)
if (!(mp = *ogmlist) )
return TRUE; /* there is nothing to check */
/* go through the list message ids. checking if the message has
complete. If the list is empty at the end, then return that
message has finished */
head = mp;
mp = mp->ms_link;
while (mp != head)
{
if ((*mp->mfunc->msgdone)(0, &(mp->id), mp->info))
{
mp = mp->ms_link;
oldmp = mp->ms_rlink;
LISTDELETE(oldmp,ms_link, ms_rlink)
msgid_free(oldmp);
}
else
mp = mp->ms_link;
}
/* mp now points to the head of the list */
if ((*mp->mfunc->msgdone)(0, &(mp->id), mp->info))
{
if (mp->ms_link == mp) /* the only one left */
{
msgid_free(mp);
*ogmlist = (struct msgid *) NULL;
return TRUE;
}
else
{
mp = mp->ms_link;
oldmp = mp->ms_rlink;
LISTDELETE(oldmp, ms_link, ms_rlink);
msgid_free(oldmp);
*ogmlist = mp; /* new head of the message id list */
}
}
return FALSE;
#else
return TRUE; /* for workstations, the write is sychronous */
#endif
}